package io.gitee.zhangbinhub.acp.cloud.loadbalancer

import com.alibaba.cloud.commons.lang.StringUtils
import com.alibaba.cloud.nacos.NacosDiscoveryProperties
import com.alibaba.cloud.nacos.loadbalancer.LoadBalancerAlgorithm
import com.alibaba.cloud.nacos.loadbalancer.ServiceInstanceFilter
import com.alibaba.cloud.nacos.util.InetIPv6Utils
import io.gitee.zhangbinhub.acp.boot.log.LogAdapter
import org.springframework.beans.factory.ObjectProvider
import org.springframework.cloud.client.ServiceInstance
import org.springframework.cloud.client.loadbalancer.DefaultResponse
import org.springframework.cloud.client.loadbalancer.EmptyResponse
import org.springframework.cloud.client.loadbalancer.Request
import org.springframework.cloud.client.loadbalancer.Response
import org.springframework.cloud.loadbalancer.core.NoopServiceInstanceListSupplier
import org.springframework.cloud.loadbalancer.core.ReactorServiceInstanceLoadBalancer
import org.springframework.cloud.loadbalancer.core.ServiceInstanceListSupplier
import reactor.core.publisher.Mono
import java.util.function.Function
import java.util.regex.Pattern
import java.util.stream.Collectors

class AcpNacosLoadBalancer(
    private val logAdapter: LogAdapter,
    private val serviceInstanceListSupplierProvider: ObjectProvider<ServiceInstanceListSupplier>,
    private val serviceId: String?,
    private val nacosDiscoveryProperties: NacosDiscoveryProperties,
    private val inetIPv6Utils: InetIPv6Utils,
    private val serviceInstanceFilters: MutableList<ServiceInstanceFilter>,
    private val loadBalancerAlgorithmMap: MutableMap<String, LoadBalancerAlgorithm>
) : ReactorServiceInstanceLoadBalancer {
    fun init() {
        val ip = nacosDiscoveryProperties.ip
        ipv6 = if (StringUtils.isNotEmpty(ip)) {
            if (Pattern.matches(IPV4_REGEX, ip)) nacosDiscoveryProperties.metadata[IPV6_KEY] else ip
        } else {
            inetIPv6Utils.findIPv6Address()
        }
    }

    private fun filterInstanceByIpType(instances: MutableList<ServiceInstance>): MutableList<ServiceInstance> {
        if (StringUtils.isNotEmpty(ipv6)) {
            val ipv6InstanceList: MutableList<ServiceInstance> = ArrayList()
            for (instance in instances) {
                if (Pattern.matches(IPV4_REGEX, instance.host)) {
                    if (StringUtils.isNotEmpty(instance.metadata[IPV6_KEY])) {
                        ipv6InstanceList.add(instance)
                    }
                } else {
                    ipv6InstanceList.add(instance)
                }
            }
            // Provider has no IPv6, should use IPv4.
            return if (ipv6InstanceList.isEmpty()) {
                instances.stream()
                    .filter { instance -> Pattern.matches(IPV4_REGEX, instance.host) }
                    .collect(Collectors.toList())
            } else {
                ipv6InstanceList
            }
        }
        return instances.stream()
            .filter { instance -> Pattern.matches(IPV4_REGEX, instance.host) }
            .collect(Collectors.toList())
    }

    override fun choose(request: Request<*>): Mono<Response<ServiceInstance>?> {
        val supplier = serviceInstanceListSupplierProvider
            .getIfAvailable { NoopServiceInstanceListSupplier() }
        return supplier.get(request).next()
            .map(Function { serviceInstances ->
                getInstanceResponse(
                    request,
                    serviceInstances
                )
            })
    }

    private fun getInstanceResponse(
        request: Request<*>,
        serviceInstances: MutableList<ServiceInstance>
    ): Response<ServiceInstance>? {
        if (serviceInstances.isEmpty()) {
            logAdapter.warn("No servers available for service: ${this.serviceId}")
            return EmptyResponse()
        }

        try {
            val clusterName = this.nacosDiscoveryProperties.clusterName

            var instancesToChoose = serviceInstances
            if (StringUtils.isNotBlank(clusterName)) {
                val sameClusterInstances = serviceInstances.stream()
                    .filter { serviceInstance ->
                        val cluster = serviceInstance.metadata["nacos.cluster"]
                        StringUtils.equals(cluster, clusterName)
                    }.collect(Collectors.toList())
                if (!sameClusterInstances.isEmpty()) {
                    instancesToChoose = sameClusterInstances
                }
            } else {
                logAdapter.warn("A cross-cluster call occurs，name = $serviceId, clusterName = $clusterName, instance = $serviceInstances")
            }
            instancesToChoose = this.filterInstanceByIpType(instancesToChoose)

            // Filter the service list sequentially based on the order number
            for (filter in serviceInstanceFilters) {
                instancesToChoose = filter.filterInstance(request, instancesToChoose)
            }

            val instance = if (loadBalancerAlgorithmMap.containsKey(serviceId)) {
                loadBalancerAlgorithmMap[serviceId]!!.getInstance(request, instancesToChoose)
            } else {
                loadBalancerAlgorithmMap[LoadBalancerAlgorithm.DEFAULT_SERVICE_ID]!!
                    .getInstance(request, instancesToChoose)
            }

            return DefaultResponse(instance)
        } catch (e: Exception) {
            logAdapter.warn("NacosLoadBalancer error", e)
            return null
        }
    }

    companion object {
        private const val IPV4_REGEX =
            "((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})(.((2(5[0-5]|[0-4]\\d))|[0-1]?\\d{1,2})){3}"

        private const val IPV6_KEY = "IPv6"

        /**
         * Storage local valid IPv6 address, it's a flag whether local machine support IPv6 address stack.
         */
        var ipv6: String? = null
    }
}
